package com.sys.midware.MQ.rocketMQ.listener;

import java.util.List;

import org.apache.log4j.Logger;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.common.message.MessageExt;

public class TestScdMessageListener extends AbstractConsumerListener {

	protected Logger logger = Logger.getLogger(getClass());
	
	@Override
	public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
			ConsumeConcurrentlyContext context) {

		ConsumeConcurrentlyStatus result = ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		for (MessageExt msg : msgs) {
			result = consumeMessage(msg);
			if(result == ConsumeConcurrentlyStatus.RECONSUME_LATER){
				break;
			}
		}

		return result;

	}
	
	/**
	 * 消费消息的模板方法。
	 *
	 * @param msg
	 * @return
	 */
	protected ConsumeConcurrentlyStatus consumeMessage(MessageExt msg){
		
		ConsumeConcurrentlyStatus result = ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		
		try {
		    
		    
//			// 检查消息来源网站
//			checkBeforeParse(msg);
//		
//			// 获取网站域名和SCD类型
//			String domain = msg.getTags();
//			String scdType = msg.getTopic();
//			statInput(domain, scdType);
//			
//			// 解析消息体
//			Map<String,String> scd = parseMsg(msg);
//			
//			// 检查消息体里数据是否合法
//			checkMessageBody(scd, domain);
//	
//			// 重新确定下scdType（-U有些类型有问题）
//			scdType = defineScdType(scd, scdType);
//			
//			// 处理消息体内容
//			result = dealMsgContent(scd, scdType, domain);
		    dualMsg(msg);		
		} catch(Exception e){
			logger.error("其他错误，下次再次消费消息", e);
			result = ConsumeConcurrentlyStatus.RECONSUME_LATER;
		}
		
		return result;
	}
	private void dualMsg(MessageExt msg){
	    try {
          //  Map<String, String> map= (Map<String, String>) JsonUtil.readJson(msg.getBody(), 0, msg.getBody().length, Map.class);
            
           // System.out.println("开始处理：" +  "tags: "  +msg.getTags()  + "map" + map); 
        } catch (Exception e) {
            
            // TODO Auto-generated catch block
            e.printStackTrace();
            
        }
	}

//	/**
//	 * 解析消息体之前做的检查工作。
//	 *
//	 * @param msg
//	 * @throws MessageInvalidException
//	 */
//	private void checkBeforeParse(MessageExt msg) throws MessageInvalidException {
//			// tags里保存的是 消息的来源网站域名
//			String domain = msg.getTags();
//			if (StringUtils.isEmpty(domain)) {
//				new MessageInvalidException("该消息没有tags[域名],不能识别SCD存放位置，已经丢弃！");
//			}
//	}
//	
//	/**
//	 * 解析消息体，返回消息体数据；
//	 *
//	 * @param msg
//	 * @return
//	 * @throws MessageParseException
//	 */
//	@SuppressWarnings("unchecked")
//	private Map<String,String> parseMsg(MessageExt msg) throws MessageParseException {
//		byte[] body = null;
//		
//		try {
//			body = msg.getBody();
//			return (Map<String, String>) JsonUtil.readJson(body, 0, body.length, Map.class);
//		} catch (Exception e) {
//			throw new MessageParseException("该消息不是合法的json格式，不能解析，data["+new String(body)+"], domain["+msg.getTags()+"]", e);
//		}
//		
//	}
//	
//	/**
//	 * 检查消息体数据是否合法；
//	 *
//	 * @param scd
//	 * @param domain
//	 * @throws MessageInvalidException
//	 */
//	private void checkMessageBody(Map<String,String> scd, String domain) throws MessageInvalidException {
//		// 检查DOCID是否存在
//		String docId = scd.get(Const.DOCID);
//		if (StringUtils.isEmpty(docId)) {
//			throw new MessageInvalidException("该消息的DOCID为空，不是合法的消息, 爬虫ip["+scd.get(Const.FROM_SERVER)	+"], domain["+domain+"]");
//		}
//	}
//	
//	/**
//	 * 统计输入域名和对应scd类型，的输入消息数目；
//	 *
//	 * @param domain
//	 * @param type
//	 */
//	private void statInput(String domain, String type){
//		// 统计接受到的消息总量
//		StatisticContainer.getInstance().get(domain, type).incrementReceiveIn();
//	}
//	
//	/**
//	 * 因爬虫端确定SCD-U类型时存在一定问题，故遇到U类型时，需要重新确定下；
//	 *
//	 * @param scd
//	 * @param scdType
//	 * @return
//	 */
//	protected String defineScdType(Map<String,String> scd, String scdType){
//		
//		if(SCD_UPDATE.equals(scdType) && StringUtils.isNotEmpty(scd.get(Const.ORIGINALPICTURE))){
//			scdType = SCD_INSERT;
//		}
//		
//		return scdType;
//	}
//	
//	/**
// 	 * 处理具体的消息体内容；
// 	 * 
//	 * @param scd
//	 * @param scdType
//	 * @param domain
//	 * @return
//	 */
//	protected ConsumeConcurrentlyStatus dealMsgContent(Map<String, String> scd, String scdType, String domain){
//		IScdDealer scdDealer = ScdDealerFactory.getScdDealer(scdType);
//		boolean result = scdDealer.dealScd(scd, scdType, domain);
//		if(result){
//			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//		}else{
//			return ConsumeConcurrentlyStatus.RECONSUME_LATER;
//		}
//	}
		
}
